# Copyright (c) 2018 Presto Labs Pte. Ltd.
# Author: xguo

from absl import flags
import datetime
import functools
import logging
import math
import numpy

from typing import List, Optional, Union

import requests
import tornado.gen
from recordclass import recordclass

import coin.flow.subscriber as flow_subscriber
from coin.base.config import Config
from coin.base.param_util import to_list
from coin.base.timestamp import get_timestamp
from coin.exchange.base.order_fill_util import FillManager
from coin.exchange.base.order_id_util import OrderIdGenerator
from coin.exchange.base.order_gateway_logger import gen_og_log_request
from coin.exchange.kr_rest.product_holders import get_holder_from_product
from coin.exchange.bitflyer_v1.kr_rest.currency import BitflyerCurrency
from coin.exchange.bitflyer_v1.rest.private_queryer import BitflyerPrivateQueryer
from coin.exchange.bitflyer_v1.ws.ws_private_client import BitflyerWsPrivateClient
from coin.exchange.bitflyer_v1.order_gateway_log import BitflyerOrderGatewayLogProvider
from coin.exchange.base.order_gateway import BaseOrderGatewayInfoMutator
from coin.proto.coin_order_gateway_pb2 import OrderGatewayConfig
from coin.exchange.bitflyer_v1.rest.async_private_client import (
    BitflyerAsyncPrivateClient,)
from coin.exchange.bitflyer_v1.kr_rest.futures_product import (
    BitflyerFuturesProduct,)
from coin.exchange.base.order_gateway import (
    Order,
    OrderGatewayBase,
    OrderGatewayStatus,
    OrderSide,
    OrderType,
)
from coin.exchange.util.rate_limit import WindowLimiter
from coin.proto.coin_query_pb2 import ProductPosition
from coin.flow.subscriber import QueueData, TopicData
from coin.proto.coin_query_pb2 import AccountBalance, AccountPosition
from coin.proto.coin_order_gateway_pb2 import OrderEvent as OrderEventProto
import coin.proto.coin_order_enums_pb2 as coin_order

flags.DEFINE_bool('bitflyer_quick_ignore', False, '')


def _time_second_to_timestamp(time_second):
  return int(time_second * 1e9)


BitflyerOrderInternal = recordclass('BitflyerOrderInternal',
                                    [
                                        'bitflyer_order_id',
                                        'in_matching_queue',
                                        'canceled_qty',
                                        'filled_qty',
                                        'ws_fill',
                                        'timestamp',
                                        'accepted_event_emitted'
                                    ])


def create_order_from_list_order_info(order_elem):
  """
   {
    "id": 138397,
    "child_order_id": "JOR20150707-084549-022519",
    "product_code": "BTC_JPY",
    "side": "SELL",
    "child_order_type": "LIMIT",
    "price": 30000,
    "average_price": 0,
    "size": 0.1,
    "child_order_state": "CANCELED",
    "expire_date": "2015-07-14T07:25:47",
    "child_order_date": "2015-07-07T08:45:47",
    "child_order_acceptance_id": "JRF20150707-084547-396699",
    "outstanding_size": 0,
    "cancel_size": 0.1,
    "executed_size": 0,
    "total_commission": 0
  }
  """
  native_symbol = order_elem['product_code']
  product = BitflyerFuturesProduct.FromStrNativeProduct(native_symbol)
  price_orig = float(order_elem['price'])
  qty_orig = float(order_elem['size'])
  order_type = getattr(OrderType, order_elem['child_order_type'])
  order_side = getattr(OrderSide, order_elem['side'])
  bitflyer_order_id = order_elem['child_order_acceptance_id']
  filled_qty = float(order_elem.get('executed_size', 0))

  ws_fill = FillManager(price_orig=price_orig,
                        qty_orig=qty_orig,
                        price_last_fill=0.0,
                        qty_last_fill=0.0,
                        price_avg_fill=0.0,
                        qty_accum_fill=0.0)

  order = Order(
      order_id=None,
      product=product,
      price=price_orig,
      qty=qty_orig,
      order_side=order_side,
      order_type=order_type,
      tag=None,
      accepted=True,
      cancel_sent=False,
      internal=BitflyerOrderInternal(
          bitflyer_order_id=bitflyer_order_id,
          filled_qty=filled_qty,
          timestamp=get_timestamp(),
          in_matching_queue=False,
          canceled_qty=0,
          ws_fill=ws_fill,
          accepted_event_emitted=False,
      ),
  )
  return order


class BitflyerGatewayInfo(BaseOrderGatewayInfoMutator):
  CurrencyType = BitflyerCurrency
  ProductType = BitflyerFuturesProduct
  BalanceType = float
  PositionType = float

  def __init__(self, currencies: List[BitflyerCurrency], products: List[BitflyerFuturesProduct]):
    super().__init__(currencies, products)
    self._market_type = gen_market_type(products)

  def update_bitflyer_balance(self, balance_message):
    self.set_zero_balances()
    for each in balance_message.each_balance:
      currency = BitflyerCurrency.FromStr(each.currency)
      self.set_balance(currency, each.total, each)

  def update_bitflyer_position(self, position_message):
    self.set_zero_positions()
    for product in self.products:
      product_position = ProductPosition(
          symbol=product.symbol,
          net_position=0.,
          net_position_contract=0,
      )
      self.set_position(product, 0., product_position)

    for each in position_message.each_position:
      product = BitflyerFuturesProduct.FromStr(each.symbol)
      self.set_position(product, each.net_position, each)

  def update_bitflyer_position_from_balance(self, balance_message):
    self.set_zero_positions()
    for each in balance_message.each_balance:
      if each.currency == "BTC":
        product = BitflyerFuturesProduct.FromStr("BTC-JPY")
        self.set_position(product, each.total)

  def balance_as_proto(self):
    return AccountBalance(exchange='Bitflyer',
                          market_type=self._market_type,
                          each_balance=self.get_notNone_balances_internal())

  def position_as_proto(self):
    return AccountPosition(exchange='Bitflyer',
                           market_type="Futures",
                           each_position=self.get_notNone_positions_internal())

  def get_products(self):
    return self._products

  def get_average_entry_price(self, product):
    return self._positions_internal[product.symbol].avg_entry_price


def gen_market_type(products):
  assert len(products) == 1
  if products[0].native_symbol == 'BTC_JPY':
    return 'Spot'
  elif products[0].native_symbol == 'FX_BTC_JPY':
    return 'Futures'
  else:
    raise NotImplementedError("not supported symbols", products)


def gen_market_subtype(products):
  assert len(products) == 1
  if products[0].native_symbol == 'BTC_JPY':
    return 'Spot'
  elif products[0].native_symbol == 'FX_BTC_JPY':
    return 'FX'
  else:
    raise NotImplementedError("not supported symbols", products)


class BitflyerOrderGateway(OrderGatewayBase):
  _exchange = 'Bitflyer'
  _market_type = None
  _api_version = None

  OGInfoType = BitflyerGatewayInfo

  def __init__(self,
               currencies,
               products: Union[List[BitflyerFuturesProduct], BitflyerFuturesProduct],
               config: Config,
               og_config=None,
               logger=None):
    super().__init__()

    self._config = config

    # 2020-03-11
    # jiangxun:muscle:  2:58 PM
    # Guys, all of our Bitflyer accounts TPS is upgraded to 600/min.
    self._special_tps = True
    if self._special_tps:
      default_og_config = OrderGatewayConfig(
          order_rate_limit=580,  # 580 queries per 1 minuts
          order_rate_limit_per=60,  # 1 minutes
      )
    else:
      default_og_config = OrderGatewayConfig(
          order_rate_limit=480,  # 480 queries per 5 minuts
          order_rate_limit_per=300,  # 5 minutes
      )

    if og_config is not None:
      default_og_config.MergeFrom(og_config)
    self._og_config = default_og_config

    self._market_type = gen_market_type(products)
    self._market_subtype = gen_market_subtype(products)

    # Orders
    self._working_orders = {}
    self._bitflyer_order_id_to_order_id = {}
    self._unknown_working_orders = {}
    self._async_private_client = None
    self._order_id_generator = OrderIdGenerator()

    self._order_complete_time = {}

    self._balance_ready = False
    self._position_ready = False
    self._orders_ready = False
    self._balance_update_count = 0
    self._position_update_count = 0
    self._orders_update_count = 0
    self._submit_in_progress_counter = 0
    self._max_imq_bitflyer_order_id = ''  # imq: in-matching-queue

    self._list_orders_received = False
    self._processed_fill_ids = set()

    self._og_info = BitflyerGatewayInfo(to_list(currencies), to_list(products))
    self._og_ready.set_status(OrderGatewayStatus.INITIALIZING)
    self._og_logger = BitflyerOrderGatewayLogProvider(gen_og_log_request(config, self))

    self._order_formatter = {
        product: get_holder_from_product(product).get_formatter()
        for product in self._og_info.products
    }

    self._ioloop = None
    self._private_queryer = None
    self._ws_private_client = None

    # Flow
    self._flow_sub = None
    self._topic_strings = ['bitflyer_rest_private_1']

    self._logger = logger or logging.getLogger(__name__)

    # The number of API queries from the same IP address will have an upper
    # limit of 500 queries per 5 minutes. Upon reaching the upper limit,
    # further API queries will be temporarily blocked. Once unblocked,
    # the upper limit will be temporarily reduced.
    window_size_ns = self._og_config.order_rate_limit_per * 1e9
    transact_limit = self._og_config.order_rate_limit

    self._logger.info('limit/window_size: %s/%s (sec)' % (transact_limit, window_size_ns / 1e9))

    self._tps_limit = WindowLimiter(window_size_ns=window_size_ns, transact_limit=transact_limit)

    # initialize full tps window on boot.
    timestamp = get_timestamp()
    start_ts = timestamp - window_size_ns
    end_ts = timestamp
    for ts_padding in numpy.arange(start_ts, end_ts, (end_ts - start_ts) / transact_limit):
      self._tps_limit.update_tps_queue(ts_padding)
      self._tps_limit.append_tps_queue(ts_padding)

  def get_tps_approval(self, timestamp=None):
    timestamp = timestamp or get_timestamp()
    self._tps_limit.update_tps_queue(timestamp)
    if self._tps_limit.get_tps_approval_cached():
      self._tps_limit.append_tps_queue(timestamp)
      return True
    else:
      logging.warning("suppressed by tps")
      return False

  def log_tps_window(self):
    self._logger.info("transact window size: %s" % self._tps_limit.get_transact_size_cached())

  def _check_ready(self):
    if self._balance_ready and self._position_ready and self._orders_ready:
      self._og_ready.set_status(OrderGatewayStatus.READY)

  def start(self, queue, ioloop, flow_sub=None):
    self._async_private_client = BitflyerAsyncPrivateClient(self._config.key_file,
                                                            market_subtype=self._market_subtype,
                                                            tps_limiter=self._tps_limit)
    self._ioloop = ioloop

    if self._special_tps:
      # Prepare private queryer
      # this consumes 60 / 2 * 6 = 180 around queries per 1min.
      self._private_queryer = BitflyerPrivateQueryer(queue,
                                                     1,
                                                     self._og_info.get_products(),
                                                     'Bitflyer',
                                                     self._async_private_client,
                                                     period=datetime.timedelta(seconds=2))
    else:
      # Prepare private queryer
      # this consumes 60 / 8 * 5 * 6 = 240 around queries per 5min.
      self._private_queryer = BitflyerPrivateQueryer(queue,
                                                     1,
                                                     self._og_info.get_products(),
                                                     'Bitflyer',
                                                     self._async_private_client,
                                                     period=datetime.timedelta(seconds=8))
    self._private_queryer.start(ioloop)

    self._prepare_ws_private_client(start=True)

    # Subscribe to flow for feed.
    self._flow_sub = flow_sub or flow_subscriber.from_queue(queue)
    self._flow_sub.subscribe(self._topic_strings[0], self.on_rest_private_msg)

  def _prepare_ws_private_client(self, start=False):
    if self._ws_private_client is not None:
      self._ws_private_client.stop()

    self._ws_private_client = BitflyerWsPrivateClient(
        self._config.key_file,
        ioloop=self._ioloop,
        notification_callback=self._notification_callback,
        close_callback=functools.partial(self._prepare_ws_private_client, True),
        on_ready=self._on_ws_private_client_ready,
    )

    if start:
      self._ws_private_client.start()

  def _on_ws_private_client_ready(self):
    channels = ['child_order_events']
    self._ws_private_client.prepare_subscription_channels(channels)

  def _notification_callback(self, data):
    self._handle_ws_orders_update(data)

  def _handle_fill_by_ws(self, order_elem, timestamp, order):
    try:
      last_fill_price = float(order_elem['price'])
      last_fill_qty = float(order_elem['size'])
      bitflyer_order_id = order_elem['child_order_acceptance_id']
    except ValueError:
      return

    if last_fill_qty > 0:
      last_fill = order.internal.ws_fill.update_by_last_fill(last_fill_price, last_fill_qty)
      if last_fill:
        self._logger.info('(WS FILL PROCESS)')
        self._publish_fill(last_fill.price_last_fill, last_fill.qty_last_fill, order)
        order.internal.ws_fill.log_last_fill(order, self._logger)
        self._og_logger.gen_order_event_and_log(order,
                                                timestamp,
                                                OrderEventProto.ORDER_FILLED,
                                                fill_price=last_fill.price_last_fill,
                                                fill_qty=last_fill.qty_last_fill,
                                                fill_source=coin_order.EXPLICIT_FILL_MESSAGE,
                                                external_order_id=bitflyer_order_id)

  def _handle_ws_orders_update(self, msg):
    timestamp = get_timestamp()
    data = msg['params']['message']
    assert isinstance(data, list)

    for order_elem in data:
      bitflyer_order_id = order_elem['child_order_acceptance_id']
      # native_symbol = order_elem['product_code']
      order_state = order_elem['event_type']

      self._max_imq_bitflyer_order_id = max(self._max_imq_bitflyer_order_id, bitflyer_order_id)

      order = self._find_order_by_bitflyer_order_id(bitflyer_order_id)
      if order:
        # ORDER, ORDER_FAILED, CANCEL, CANCEL_FAILED, EXECUTION, EXPIRE
        # this ORDER may come faster than submit's response..(but it would be ignored safely)
        # proceed fill
        if order_state == 'EXECUTION' and bitflyer_order_id not in self._processed_fill_ids:
          self._handle_fill_by_ws(order_elem, timestamp, order=order)
        if order.internal.ws_fill.fully_filled:
          order_state = 'COMPLETE'

        if order_state in ('COMPLETE', 'ORDER_FAILED', 'CANCEL', 'EXPIRE'):
          if order_state == 'COMPLETE':
            self._logger.info('(WS) COMPLETE ORDER!')
          elif order_state == 'CANCEL':
            self._og_logger.gen_order_event_and_log(order,
                                                    timestamp,
                                                    OrderEventProto.CANCEL_CONFIRMED)
          elif order_state == 'EXPIRED':
            self._og_logger.gen_order_event_and_log(order, timestamp, OrderEventProto.ORDER_EXPIRED)
          elif order_state == 'ORDER_FAILED':
            self._og_logger.gen_order_event_and_log(order,
                                                    timestamp,
                                                    OrderEventProto.ORDER_REJECTED)

        else:  # this is active orders(partial fill)
          if not order.internal.in_matching_queue:
            self._logger.debug('From Ws Msg, In matching queue: %f',
                               (get_timestamp() - order.internal.timestamp) / 1.e9)
          order.internal.in_matching_queue = True
          if not order.internal.accepted_event_emitted:
            order.internal.accepted_event_emitted = True
            self._og_logger.gen_order_event_and_log(order,
                                                    timestamp,
                                                    OrderEventProto.ORDER_ACCEPTED)

    self._check_ready()

  def stop(self):
    # Unsubscribe topic
    self._flow_sub.unsubscribe(self._topic_strings[0], self.on_rest_private_msg)
    self._private_queryer.stop()
    self._async_private_client.stop()
    self._ws_private_client.stop()

  def _find_orders_by_bitflyer_order_id(self, bitflyer_order_id_list):
    res = []
    for bitflyer_order_id in bitflyer_order_id_list:
      order_id = self._bitflyer_order_id_to_order_id.get(bitflyer_order_id)
      if order_id and order_id in self._working_orders:
        res.append(self._working_orders[order_id])
    return res

  def _find_order_by_bitflyer_order_id(self, bitflyer_order_id):
    order_id = self._bitflyer_order_id_to_order_id.get(bitflyer_order_id)
    if order_id and order_id in self._working_orders:
      return self._working_orders[order_id]
    else:
      return None

  def _handle_rest_private_balance(self, message):
    self._og_info.update_bitflyer_balance(message)
    self._og_logger.write_balance(self._og_info.balance_as_proto())
    self._balance_update_count += 1
    self._balance_ready = self._balance_update_count > 3
    self._check_ready()

    if self._market_subtype == "Spot":
      self._og_info.update_bitflyer_position_from_balance(message)
      self._og_logger.write_position(self._og_info.position_as_proto())
      self._position_update_count += 1
      self._position_ready = self._position_update_count > 3
      self._check_ready()
    elif self._market_subtype == "FX":
      pass
    else:
      raise NotImplementedError()

  def _handle_rest_private_position(self, message):
    if self._market_subtype == "Spot":
      pass
    elif self._market_subtype == "FX":
      self._og_info.update_bitflyer_position(message)
      self._og_logger.write_position(self._og_info.position_as_proto())
      self._position_update_count += 1
      self._position_ready = self._position_update_count > 3
      self._check_ready()
    else:
      raise NotImplementedError()

  def _handle_fill(self, order_elem, timestamp, order):
    order_state = order_elem['child_order_state']
    if order_state not in ('COMPLETED', 'CANCELED', 'EXPIRED', 'REJECTED'):
      return

    bitflyer_order_id = order_elem['child_order_acceptance_id']
    if bitflyer_order_id in self._processed_fill_ids:
      return

    self._processed_fill_ids.add(bitflyer_order_id)
    if not self._list_orders_received:
      return

    # native_symbol = order_elem['product_code']
    # order_side_str = order_elem['side'].upper()
    # order_side = None
    # if order_side_str == 'BUY':
    #   order_side = OrderSide.BUY
    # elif order_side_str == 'SELL':
    #   order_side = OrderSide.SELL

    try:
      avg_fill_price = float(order_elem['average_price'])
      fill_qty = float(order_elem['executed_size'])
    except ValueError:
      return

    if fill_qty > 0:
      # product = BitflyerFuturesProduct.FromStrNativeProduct(native_symbol)
      if order is None:
        try:
          order = create_order_from_list_order_info(order_elem)
        except (ValueError, AttributeError, KeyError):
          self._logger.error('Create order failed! order_elem: %s', order_elem)

      # subtract from ws fill.
      last_fill = order.internal.ws_fill.update_by_accum_fill(avg_fill_price, fill_qty)
      if last_fill:
        self._logger.info('(REST FILL PROCESS)')
        self._publish_fill(last_fill.price_last_fill, last_fill.qty_last_fill, order)
        order.internal.ws_fill.log_last_fill(order, self._logger)
        self._og_logger.gen_order_event_and_log(order,
                                                timestamp,
                                                OrderEventProto.ORDER_FILLED,
                                                fill_price=last_fill.price_last_fill,
                                                fill_qty=last_fill.qty_last_fill,
                                                fill_source=coin_order.IMPLIED_FROM_DEAD,
                                                external_order_id=bitflyer_order_id)

  # This handler method is called regularly.
  def _handle_rest_private_list_orders(self, message, timestamp):
    assert isinstance(message, list)
    unknown_working_orders = {}

    retrieved_active_order_ids = set()
    for order_elem in message:
      bitflyer_order_id = order_elem['child_order_acceptance_id']

      # native_symbol = order_elem['product_code']
      # TODO(xguo): if not trading symbol, continue.

      order_type = order_elem['child_order_type']
      order_state = order_elem['child_order_state']
      price = order_elem['price']
      orig_qty = order_elem['size']
      cancel_qty = order_elem['cancel_size']
      executed_qty = order_elem['executed_size']
      outstanding_qty = order_elem['outstanding_size']
      assert (price > 0 and order_type != "MARKET") or (order_type == "MARKET")
      assert orig_qty > 0
      assert cancel_qty >= 0
      assert executed_qty >= 0
      assert (executed_qty + cancel_qty) <= (orig_qty + 0.001), (executed_qty, cancel_qty, orig_qty)
      assert math.isclose(outstanding_qty + executed_qty + cancel_qty, orig_qty)

      if order_elem['side'] == 'BUY':
        # order_side = OrderSide.BUY
        pass
      elif order_elem['side'] == 'SELL':
        # order_side = OrderSide.SELL
        pass
      else:
        raise ValueError('Invalid order side')

      self._max_imq_bitflyer_order_id = max(self._max_imq_bitflyer_order_id, bitflyer_order_id)

      order = self._find_order_by_bitflyer_order_id(bitflyer_order_id)
      if order:
        assert math.isclose(order.price, price)
        assert math.isclose(order.qty, orig_qty)
        if math.isclose(outstanding_qty, 0.0) or order_state != "ACTIVE":
          if order_state == 'COMPLETED':
            pass
          elif order_state == 'CANCELED':
            self._og_logger.gen_order_event_and_log(order,
                                                    timestamp,
                                                    OrderEventProto.CANCEL_CONFIRMED)
          elif order_state == 'EXPIRED':
            self._og_logger.gen_order_event_and_log(order, timestamp, OrderEventProto.ORDER_EXPIRED)
          elif order_state == 'REJECTED':
            self._og_logger.gen_order_event_and_log(order,
                                                    timestamp,
                                                    OrderEventProto.ORDER_REJECTED)
          del self._working_orders[order.order_id]

        else:
          assert not math.isclose(outstanding_qty, 0.0), outstanding_qty
          retrieved_active_order_ids.add(order.order_id)
          if not order.internal.in_matching_queue:
            self._logger.debug('In matching queue: %f',
                               (get_timestamp() - order.internal.timestamp) / 1.e9)
          order.internal.in_matching_queue = True
          if not order.internal.accepted_event_emitted:
            order.internal.accepted_event_emitted = True
            self._og_logger.gen_order_event_and_log(order,
                                                    timestamp,
                                                    OrderEventProto.ORDER_ACCEPTED)

      elif self._submit_in_progress_counter == 0:
        if order_state not in ('COMPLETED', 'CANCELED', 'EXPIRED', 'REJECTED'):
          unknown_working_orders[bitflyer_order_id] = dict(order_elem)
          if bitflyer_order_id not in self._unknown_working_orders:
            self._logger.warning('Unknown working order: %s', str(order_elem))

      self._handle_fill(order_elem, timestamp, order=order)

    self._list_orders_received = True

    for order_id in list(self._working_orders.keys()):
      if (self._working_orders[order_id].internal.in_matching_queue
          and order_id not in retrieved_active_order_ids):
        self._logger.info('Deleting order [%s] due to the absence from the order list', order_id)
        del self._working_orders[order_id]

    self._unknown_working_orders = unknown_working_orders

    self.cleanup_fuckedup_orders()
    self._orders_update_count += 1
    self._orders_ready = self._orders_update_count > 3
    self._check_ready()

  def on_rest_private_msg(self, record, queue_data: QueueData, topic_data: TopicData):
    assert topic_data.topic_string
    assert queue_data.queue_name

    message = record.data

    query_type = message['query']
    if query_type == 'query_balance':
      self._handle_rest_private_balance(message['data'])
    elif query_type == 'query_position':
      self._handle_rest_private_position(message['data'])
    elif query_type == 'query_list_orders':
      self._handle_rest_private_list_orders(message['data'], record.timestamp)
    else:
      raise ValueError('Unknown query: %s' % query_type)

  def cleanup_fuckedup_orders(self):
    # order_id_list contains order_id of recently completed orders.
    ignore_list = []
    timestamp = get_timestamp()

    for order_id, order in self._working_orders.items():
      if not order.accepted:
        continue

      if not order.internal.in_matching_queue:
        elapsed_time = (timestamp - order.internal.timestamp) / 1e9
        if flags.FLAGS.bitflyer_quick_ignore:
          if order.internal.bitflyer_order_id < self._max_imq_bitflyer_order_id:
            cond = elapsed_time > 3.
          else:
            cond = elapsed_time > 5.
        else:
          if order.internal.bitflyer_order_id < self._max_imq_bitflyer_order_id:
            cond = elapsed_time > 5.
          else:
            cond = elapsed_time > 60.
        if cond:
          self._logger.info('Order [%s] is ignored!', order_id)
          self.cancel(order_id)  # Try to cancel order.
          ignore_list.append(order_id)

    for order_id in ignore_list:
      order = self._working_orders[order_id]
      self._og_logger.gen_flush_unacked_order_event_and_log(order.product.symbol,
                                                            order.order_id,
                                                            timestamp)
      del self._working_orders[order_id]

  def submit_impl(
      self,
      product: BitflyerFuturesProduct,
      price: float,
      qty: float,
      order_side: OrderSide,
      order_type: OrderType = OrderType.LIMIT,  # LIMIT
      order_id: Optional[str] = None,
      tag=None,
      minute_to_expire=1):
    assert isinstance(product, BitflyerFuturesProduct)
    assert self.is_ready()
    assert price > 0
    assert qty > 0
    assert order_type == OrderType.LIMIT

    timestamp = get_timestamp()
    order_id = order_id or self._order_id_generator.gen_order_id()
    self._og_logger.gen_order_request_and_order_event_and_log(product.symbol,
                                                              price,
                                                              qty,
                                                              order_side,
                                                              order_type,
                                                              order_id,
                                                              timestamp)

    ws_fill = FillManager(price_orig=price,
                          qty_orig=qty,
                          price_last_fill=0.0,
                          qty_last_fill=0.0,
                          price_avg_fill=0.0,
                          qty_accum_fill=0.0)

    order = Order(order_id=order_id,
                  product=product,
                  price=price,
                  qty=qty,
                  order_side=order_side,
                  order_type=order_type,
                  tag=tag,
                  accepted=False,
                  cancel_sent=False,
                  internal=BitflyerOrderInternal(timestamp=timestamp,
                                                 bitflyer_order_id=None,
                                                 in_matching_queue=False,
                                                 canceled_qty=None,
                                                 filled_qty=None,
                                                 ws_fill=ws_fill,
                                                 accepted_event_emitted=False))
    self._working_orders[order_id] = order

    self._ioloop.add_callback(self._submit, timestamp, order_id, order, minute_to_expire)

  @tornado.gen.coroutine
  def _submit(self, timestamp, order_id, order, minute_to_expire):
    if order.order_side == OrderSide.BUY:
      side = 'BUY'
    elif order.order_side == OrderSide.SELL:
      side = 'SELL'
    else:
      raise ValueError('Unknown side: %s' % order.order_side)

    order_formatter = self._order_formatter[order.product]
    fmt_price = order_formatter.format_price(order.price)
    fmt_qty = order_formatter.format_qty(order.qty)

    try:
      self._submit_in_progress_counter += 1
      submit_ts = get_timestamp()
      if self.get_tps_approval():
        update = yield self._async_private_client.async_query_submit(order.product,
                                                                     side,
                                                                     fmt_price,
                                                                     fmt_qty,
                                                                     minute_to_expire)
      else:
        update = None
    except Exception as e:
      update = None
      self._logger.exception('Submit order error. order_time=%d submit_time=%d error_time=%d\n%s',
                             order.internal.timestamp,
                             submit_ts,
                             get_timestamp(),
                             str(e))
    finally:
      self._submit_in_progress_counter -= 1

    self.log_tps_window()

    # After getting response.
    if update is not None:
      self._logger.info('Submit order success. order_time=%d submit_time=%d success_time=%d',
                        order.internal.timestamp,
                        submit_ts,
                        get_timestamp())

      bitflyer_order_id = update.native_msg['child_order_acceptance_id']
      # Bitflyer accept isn't really accepted.
      order.accepted = True
      order.internal.bitflyer_order_id = bitflyer_order_id
      self._bitflyer_order_id_to_order_id[bitflyer_order_id] = order_id
      self._logger.debug('Submit order succeeded! %s' % order_id)
      res_timestamp = _time_second_to_timestamp(update.res_timestamp)
      self._og_logger.gen_order_response_and_order_event_and_log(order.product.symbol,
                                                                 True,
                                                                 order_id,
                                                                 res_timestamp)
    else:
      del self._working_orders[order_id]
      self._logger.debug('Submit order rejected! %s' % order_id)
      self._og_logger.gen_order_response_and_order_event_and_log(order.product.symbol,
                                                                 False,
                                                                 order_id,
                                                                 get_timestamp())

  def cancel_impl(self, *args, **kwargs):
    self._ioloop.add_callback(self._cancel, *args, **kwargs)

  @tornado.gen.coroutine
  def _cancel(self, order_id, *, ignore_error=False):
    assert self.is_ready()
    if order_id not in self._working_orders:
      self._logger.error('No such order. order_id=%s' % order_id)
      return

    if not self._working_orders[order_id].accepted:
      self._logger.error('Order has not been accepted: %s' % order_id)
      return

    if self._working_orders[order_id].cancel_sent:
      self._logger.error('Order has already cancel_sent: %s' % order_id)
      return

    timestamp = get_timestamp()
    order = self._working_orders[order_id]
    bitflyer_order_id = order.internal.bitflyer_order_id
    order.cancel_sent = True
    self._og_logger.gen_cancel_request_and_order_event_and_log(order.product.symbol,
                                                               order_id,
                                                               timestamp)

    try:
      if self.get_tps_approval():
        update = yield self._async_private_client.async_query_cancel(order.product,
                                                                     bitflyer_order_id)
      else:
        update = None
    except Exception as e:
      update = None
      self._logger.exception('Cancel order error! %s' % e)
    self.log_tps_window()

    if update is not None and update.status == requests.codes.ok:
      self._logger.info('Cancel order succeeded! %s' % order_id)
      self._og_logger.gen_cancel_response_and_log(order.product.symbol, True, order_id)
      # Bitflyer cancel does not really make the order canceled.
      # del self._working_orders[order_id]
    else:
      self._logger.info('Cancel order rejected! %s' % order_id)
      order.cancel_sent = False
      self._og_logger.gen_cancel_response_and_log(order.product.symbol, False, order_id)

  def cancel_multiple_impl(self, order_ids, *, ignore_error=False):
    for order_id in order_ids:
      self.cancel(order_id)

  # Bitflyer provides cancel all API, no OrderEvent is emitted. Use this
  # method for cancelling orders from previous session.
  def cancel_all_impl(self, *, ignore_error=False):
    product_list = set(self._og_info.products)
    for order in self._working_orders.values():
      product_list.add(order.product)

    for product in product_list:
      if self.get_tps_approval():
        self._async_private_client.async_query_cancel_product(product)
    self.log_tps_window()

  def cancel_unknown_working_orders(self):
    if not self._unknown_working_orders:
      return

    self._logger.info('Cancel all unknown working orders (%d).', len(self._unknown_working_orders))
    for bitflyer_order_id, order_elem in self._unknown_working_orders.items():
      native_symbol = order_elem['product_code']
      product = BitflyerFuturesProduct.FromStrNativeProduct(native_symbol)
      if self.get_tps_approval():
        self._ioloop.add_callback(self._async_private_client.async_query_cancel,
                                  product,
                                  bitflyer_order_id)
    self.log_tps_window()

  def get_working_order(self):
    return self._working_orders.values()

  def has_pending_order(self, product):
    for order in self._working_orders.values():
      if order.product == product and not order.accepted:
        return True
    return False

  def get_average_entry_price(self, product):
    return self._og_info.get_average_entry_price(product)
